package org.zjx.server;

import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import org.zjx.core.ClusterConfig;
import org.zjx.core.SocketConfig;
import org.zjx.core.SocketRemoteInvoker;
import org.zjx.handler.CommandHandler;
import org.zjx.handler.CommandHandlerFactory;
import org.zjx.message.*;

import java.io.*;
import java.net.Socket;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

@Getter
@Setter
public class SocketServer extends BaseSocketServer {
    private final ClusterCommandProcessor commandProcessor;
    private final SessionManager sessionManager;
    private final ScheduledExecutorService heartbeatChecker;

    public SocketServer(SocketConfig config, ClusterConfig clusterConfig) {
        super(config);
        this.sessionManager = SessionManager.getInstance();
        this.heartbeatChecker = Executors.newScheduledThreadPool(1);

        this.commandProcessor = (clusterConfig != null) ?
                new ClusterCommandProcessor(clusterConfig, new SocketRemoteInvoker(config)) : null;
    }

    @SneakyThrows
    @Override
    protected void handleConnection(Socket clientSocket) {
        new Thread(new ClientHandler(clientSocket)).start();
    }

    @Override
    public void close() throws IOException {
        // 关闭心跳检测线程池
        if (heartbeatChecker != null) {
            heartbeatChecker.shutdownNow();
            try {
                if (!heartbeatChecker.awaitTermination(3, TimeUnit.SECONDS)) {
                    logWarn("Heartbeat checker did not terminate in time");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        // 关闭所有客户端连接
        sessionManager.getAllSessions().values().forEach(session -> {
            closeQuietly(session.getSocket());
        });

        super.close();
    }

    private class ClientHandler implements Runnable {
        private final Socket socket;
        private final BufferedReader in;
        private final PrintWriter out;
        private final ClientSession session;
        private volatile long lastActivity = System.currentTimeMillis();
        private volatile boolean connectionActive = true;
        private ScheduledFuture<?> heartbeatFuture;

        public ClientHandler(Socket socket) throws IOException {
            this.socket = socket;
            this.in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            this.out = new PrintWriter(socket.getOutputStream(), true);
            this.session = sessionManager.createSession(socket, out);
        }

        @Override
        public void run() {
            logInfo("Client connected: " + session.getSessionId());

            // 启动心跳检测（如果配置）
            if (config.getHeartbeatTimeout() > 0) {
                heartbeatFuture = startHeartbeatMonitor();
            }

            try {
                processClientRequests();
            } catch (IOException e) {
                if (running && connectionActive) {
                    logError("Connection error", e);
                }
            } finally {
                cleanupConnection();
            }
        }

        private void processClientRequests() throws IOException {
            String inputLine;
            while (running && connectionActive && (inputLine = in.readLine()) != null) {
                lastActivity = System.currentTimeMillis();
                processMessage(inputLine);
            }
        }

        private void processMessage(String messageJson) {
            try {
                Message message = Message.parse(messageJson);
                switch (message.getType()) {
                    case HEARTBEAT:
                        handleHeartbeat((HeartbeatMessage) message);
                        break;
                    case COMMAND:
                        handleCommand((CommandMessage) message);
                        break;
                    default:
                        sendError("Unsupported message type");
                }
            } catch (Exception e) {
                logError("Message processing error", e);
                sendError("Invalid message format");
            }
        }

        private void handleHeartbeat(HeartbeatMessage message) {
            // 更新心跳时间并回复
            lastActivity = System.currentTimeMillis();
            sendMessage(new HeartbeatMessage());
        }

        private void handleCommand(CommandMessage cmd) {
            // 特殊处理注册命令
            if ("REGISTER".equals(cmd.getCommand())) {
                handleRegistration(cmd);
                return;
            }

            // 普通命令处理
            ResponseMessage response = (commandProcessor != null) ?
                    commandProcessor.processCommand(cmd) :
                    executeLocally(cmd);

            sendMessage(response);
        }

        private void handleRegistration(CommandMessage cmd) {
            String clientId = (String) cmd.getParams().get("clientId");
            if (clientId != null && !clientId.isEmpty()) {
                session.setClientId(clientId);
                sessionManager.bindClientToSession(clientId, session.getSessionId());

                Optional<CommandHandler> commandHandler = CommandHandlerFactory.getHandler("REGISTER");
                if (commandHandler.isPresent()) {
                    sendMessage(commandHandler.get().handle(cmd));
                } else {
                    sendMessage(ResponseMessage.success(cmd, "Registered successfully"));
                }
            } else {
                sendMessage(ResponseMessage.error(cmd, "Invalid client ID"));
            }
        }

        private ResponseMessage executeLocally(CommandMessage command) {
            return CommandHandlerFactory.getHandler(command.getCommand())
                    .map(handler -> handler.handle(command))
                    .orElse(ResponseMessage.error(command, "Unsupported command"));
        }

        private ScheduledFuture<?> startHeartbeatMonitor() {
            return heartbeatChecker.scheduleAtFixedRate(() -> {
                // 双重检查确保连接有效
                if (!connectionActive || socket.isClosed() || !session.isActive()) {
                    return;
                }

                long inactiveTime = System.currentTimeMillis() - lastActivity;
                if (inactiveTime > config.getHeartbeatTimeout()) {
                    logWarn("Heartbeat timeout, closing connection: " + session.getSessionId());

                    // 原子性关闭操作
                    synchronized (this) {
                        if (connectionActive) {
                            connectionActive = false;
                            closeQuietly(socket);
                            session.setActive(false);
                        }
                    }
                }
            }, 10, 10, TimeUnit.SECONDS);
        }

        private void cleanupConnection() {
            // 取消心跳检测任务
            if (heartbeatFuture != null) {
                heartbeatFuture.cancel(true);
            }

            // 标记连接不活跃
            connectionActive = false;
            sessionManager.removeSession(session.getSessionId());

            // 安全关闭资源
            closeQuietly(in);
            closeQuietly(out);
            closeQuietly(socket);

            logInfo("Client disconnected: " + session.getSessionId());
        }

        private void sendError(String error) {
            sendMessage(ResponseMessage.error(null, error));
        }

        private void sendMessage(Message message) {
            if (connectionActive) {
                out.println(message.toJsonString());
            }
        }
    }

    // 消息推送接口
    public void pushToClient(String clientId, Message message) {
        ClientSession session = sessionManager.getSessionByClientId(clientId);
        if (session != null && session.isActive()) {
            session.sendMessage(message);
        }
    }
}